package org.spark.api.util;

import com.google.common.collect.ImmutableList;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.spark.api.annotation.HBaseColumn;
import org.spark.api.annotation.HBaseTable;
import org.spark.api.entity.Author;
import org.spark.api.util.model.HBaseClassMapping;
import org.spark.api.util.model.HBaseColumnMapping;
import org.spark.api.util.model.HBaseTableMapping;
import java.beans.BeanInfo;
import java.beans.IntrospectionException;
import java.beans.Introspector;
import java.beans.PropertyDescriptor;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

/**
 * @author chenwy
 */
public class ReflectHBaseUtils {

    public static PropertyDescriptor[] getPropertiesHelper(Class clazz, boolean read, boolean write) {

        try {
            BeanInfo info = Introspector.getBeanInfo(clazz, Object.class);
            PropertyDescriptor[] all = info.getPropertyDescriptors();
            if (read && write) {
                return all;
            }
            List properties = new ArrayList(all.length);
            for (int i = 0; i < all.length; i++) {
                PropertyDescriptor pd = all[i];
                boolean allowRead = read && pd.getReadMethod() != null;
                boolean allowWrite = write && pd.getWriteMethod() != null;
                if (allowRead || allowWrite) {
                    properties.add(pd);
                }
            }
            return (PropertyDescriptor[]) properties.toArray(new PropertyDescriptor[properties.size()]);
        } catch (IntrospectionException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 解析HBase表映射
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> HBaseColumnMapping[] resolveHBaseTableMapping(Class<T> clazz) {
        ImmutableList.Builder<HBaseColumnMapping> builder = ImmutableList.builder();
        try {
            PropertyDescriptor[] descriptors = getPropertiesHelper(clazz, true, true);
            clazz.isAnnotationPresent(HBaseTable.class);
            for (PropertyDescriptor descriptor : descriptors) {
                String name = descriptor.getName();
                Field field = clazz.getDeclaredField(name);
                if (field.isAnnotationPresent(HBaseColumn.class)) {
                    HBaseColumn hBaseColumn = field.getAnnotation(HBaseColumn.class);
                    HBaseColumnMapping columnFilter = new HBaseColumnMapping();
                    columnFilter.setFamily(hBaseColumn.family().getBytes());
                    columnFilter.setQualifier(hBaseColumn.column().getBytes());
                    columnFilter.setUseMsgPack(hBaseColumn.useMsgPack());
                    columnFilter.setValueClass(field.getType());
                    columnFilter.setField(field);
                    builder.add(columnFilter);
                }
            }
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        }
        ImmutableList<HBaseColumnMapping> list = builder.build().asList();
        HBaseColumnMapping[] columnFilters = list.toArray(new HBaseColumnMapping[]{});
        HBaseTableMapping.put(clazz, columnFilters);
        return columnFilters;
    }


    /**
     * 获取class
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> HBaseClassMapping getHBaseClassMapping(Class<T> clazz){
        HBaseClassMapping classMapping = HBaseTableMapping.get(clazz);
        if(classMapping == null){
            ReflectHBaseUtils.resolveHBaseTableMapping(clazz);
            classMapping = HBaseTableMapping.get(clazz);
        }
        return classMapping;
    }


    public static <T> T parseObject(Result result, Class<T> clazz) {
        try {
            if (!result.isEmpty()) {
                T instance = clazz.newInstance();
                CellScanner scanner = result.cellScanner();
                HBaseClassMapping mapping = HBaseTableMapping.get(clazz);
                while (scanner.advance()) {
                    Cell cell = scanner.current();
                    String family = new String(CellUtil.cloneFamily(cell));
                    String qualifier = new String(CellUtil.cloneQualifier(cell));
                    byte[] value = CellUtil.cloneValue(cell);
                    if (mapping.containsKey(family + qualifier)) {
                        HBaseColumnMapping filter = mapping.get(family + qualifier);
                        Object ob = ColumnValueConverter.converter(value, filter.getValueClass(), filter.isUseMsgPack());
                        Field field = filter.getField();
                        field.setAccessible(true);
                        field.set(instance, ob);
                    }
                }
                return instance;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public static void main(String[] args) {
        HBaseColumnMapping[] columnFilters = resolveHBaseTableMapping(Author.class);
    }

}
